#include <stdio.h>
#include <unistd.h>
#include <pthread.h>
#include <string.h>
#include <log.h>
#include <timer.h>

#include "scu.h"
#include "scc.h"
#include "sc.h"

#define BUFSIZE 65536
#define CHUNKSIZE 1024
#define TRIALS  ((1024*1024*1024)/CHUNKSIZE)
#define READERS 1

scu_buf_t bufa, bufb;
int running[READERS];

void *writer(void*);
void *reader(void*);

void mydatacallback(scc_hdr_t *hdr);
unsigned long int myquerycallback(int buffer);

extern table_t scc_trans_procs;

int main(void)
{
  int i, init;
  pthread_t w, r[READERS];
  int argc = 6;
  char *argv[6] = {"./transtest", "-m", "machines.xml", "-n", "localhost", "-c"};

  scInit(argc, argv);

  memset(running, 0, READERS*sizeof(int));
  
  bufa = scuBufferCreate(BUFSIZE, SCU_BUF_MAX_READERS, NULL);
  bufb = scuBufferCreate(BUFSIZE, SCU_BUF_MAX_READERS, NULL);

  tbldel(&scc_trans_procs, 1, 1);
  sccTransportRegister(1, mydatacallback, myquerycallback);

  printf("Starting threads\n");
  pthread_create(&w, NULL, writer, NULL);
  
  for (i=0; i < READERS; i++)
    pthread_create(&r[i], NULL, reader, (void*)i);

  do
  {
    init = 1;
    for (i=0; i < READERS; i++)
      init = init * running[i];
  } while (!init);  
  scuBufferUnlock(&bufb);

  printf("Creating transport\n");
  sccTransportCreate(&bufa, 1, 1, 0);
  scuBufferUnlock(&bufa);

  printf("Benchmarking\n");
  pthread_join(w, NULL);
  for (i=0; i < READERS; i++)
    pthread_join(r[i], NULL);

  return 0;
}

void *writer(void* dummy)
{
  char *data;
  int i;

  logRegisterThread("writer");

  printf("Starting writer\n");
  for (i=0; i < TRIALS; i++)
  {
    scuBufferAllocate(&bufa, (void**)&data, CHUNKSIZE, SCU_BUF_BLOCK_ALL);
    scuBufferReleaseAllocated(&bufa, CHUNKSIZE);
  }

  return NULL;
}

void *reader(void* vid)
{
  char *data;
  char buf[PATH_MAX];
  scu_bufptrid_t r;
  int i=0;
  int id = (int)vid;
  tmr_t timer = tmrInit();
  
  sprintf(buf, "reader%03d", id);
  logRegisterThread(buf);

  r = scuBufferAddReader(&bufb);

  running[id] = 1;

  printf("Starting reader %d\n", id);
  
  tmrStart(&timer);
  for (i=0; i < TRIALS; i++)
  {
    scuBufferPeek(&bufb, r, (void**)&data, CHUNKSIZE, SCU_BUF_BLOCK_ALL);
    scuBufferReleasePeeked(&bufb, r, CHUNKSIZE);
  }
  tmrStop(&timer);

  printf("Speed was %f MB/s\n", ((double)TRIALS)*CHUNKSIZE/tmrGetValueInUs(&timer));

  return NULL;
}

void mydatacallback(scc_hdr_t *hdr)
{
  scc_datamsg_t *msg = (scc_datamsg_t*)hdr;
  scu_buf_t *buf = &bufb;
    
  switch (hdr->type)
  {
    case SCC_MSGTYPE_BUFFERDATA:
      if (scuBufferSpaceAvailable(buf) < hdr->size-(sizeof(scc_datamsg_t)-sizeof(scc_hdr_t)))
        printf("Incoming communication of size %ld for buffer %d (%d available) would block\n",
                hdr->size-(sizeof(scc_datamsg_t)-sizeof(scc_hdr_t)), msg->buffer, scuBufferSpaceAvailable(buf));

      scuBufferWrite(buf, msg->data, hdr->size-(sizeof(scc_datamsg_t)-sizeof(scc_hdr_t)),
		     SCU_BUF_BLOCK_ALL);
      break;
    case SCC_MSGTYPE_BUFFERDRY:
      scuBufferFinalize(buf);
      break;
    default:
      printf("Unknown message type %d\n", hdr->type);
  }
}

unsigned long int myquerycallback(int buffer)
{
  return scuBufferSpaceAvailable(&bufb);
}
